package com.zy.smarthome.handler;

import com.zy.smarthome.common.content.RedisKey;
import com.zy.smarthome.common.util.StrUtil;
import com.zy.smarthome.frame.FrameStaticData;
import com.zy.smarthome.frame.util.FrameDataDeal;
import com.zy.smarthome.frame.util.Parse2;
import com.zy.smarthome.service.ThreadPoolService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.TimeUnit;


/**
 * NettyServer通道适配器
 * <p>
 *
 * @author wjming
 * @Sharable //共享
 * @date 2018/7/9 - 上午9:39
 */
@Component
@Sharable
@Slf4j(topic = "channelDeal")
public class ServerChannelHandlerAdapter extends ChannelHandlerAdapter {

    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private ThreadPoolService threadPoolService;

    @Autowired
    private Parse2 parse2;
    /**
     * 通道组
     */
    private ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);


    /**
     * 通道注册
     * 1:把通道注册到 channelGroup中，方便rabbit通过ChannelId寻找
     * 2:上报就立即发送5010登陆帧
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) {
        try {
            channelGroup.add(ctx.channel());
            ctx.writeAndFlush(FrameDataDeal.dataToByte(FrameStaticData.get5010HeartBeatFrame()));
            log.info("{} -> [注册总通道数量] {}", this.getClass().getName(), channelGroup.size());
        } catch (Exception e) {
            log.error("{} -> [注册异常] {} ->", this.getClass().getName(), ctx.channel().id().asShortText());
        }

    }

    /**
     * 服务器接收到消息时进行进行的处理
     * 1.获取长链接ID
     * 2.数据转化位16进制
     * 3.数据校验与发送
     * 4.删除redis未连接次数短连接计数缓存
     *
     * @param ctx channelHandlerContext
     * @param msg msg
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            String key = RedisKey.heartbeat(ctx.channel().id().asLongText());
            String longId = ctx.channel().id().asLongText();
            //获取16进制数据
            String hexStr = StrUtil.getHexStr(msg);
            parse2.checkAndSend(hexStr, longId, ctx, channelGroup);
            redisTemplate.delete(key);
        } catch (Exception e) {
            log.error("{} -> [网关回应数据解析异常] {} ->", this.getClass().getName(), channelGroup.size());
        } finally {
            ReferenceCountUtil.release(msg);
        }

    }


    /**
     * 触发器
     *
     * @param ctx channelHandlerContext
     * @param evt
     * @throws Exception exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        String key = RedisKey.heartbeat(ctx.channel().id().asLongText());
        try {
            log.info("{} -> [已经有6分钟中没有接收到客户端的消息了]", this.getClass().getName());
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
                if (idleStateEvent.state() == IdleState.READER_IDLE) {
                    //从redis中获取上报信息
                    Object channelCount = redisTemplate.opsForValue().get(key);
                    if (channelCount == null) {
                        redisTemplate.opsForValue().set(key, "1", 6, TimeUnit.MINUTES);
                        //发送心跳包下去
                        ctx.writeAndFlush(FrameDataDeal.dataToByte(FrameStaticData.get5010HeartBeatFrame()));
                    } else {
                        int count = Integer.parseInt(channelCount.toString());
                        if (count > 2) {
                            log.info("{} -> [释放不活跃通道] {}", this.getClass().getName(), key);
                            channelGroup.remove(ctx.channel());
                            ctx.channel().close();
                            redisTemplate.delete(key);
                        } else {
                            redisTemplate.opsForValue().set(key, (count + 1) + "", 6, TimeUnit.MINUTES);
                            //发送心跳包下去
                            ctx.writeAndFlush(FrameDataDeal.dataToByte(FrameStaticData.get5020RepFrame()));
                        }
                    }

                }
            } else {
                super.userEventTriggered(ctx, evt);
            }
        } catch (Exception e) {
            log.error("触发器异常 [{}] ", e.getCause());
        }

    }

    /**
     * @Description:通道被异常关闭，1、有可能是网关断线了 Connection reset by peer
     * @param: ctx
     * @param: cause
     * @return: void
     * @author:wjming
     * @Date:2021/2/5 8:57
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        try {
            log.error("{} -> [连接异常] {}通道异常，异常原因：{}", this.getClass().getName(),
                    ctx.channel().id(), cause.getMessage());
            channelGroup.remove(ctx.channel());
            ctx.close();
            log.info("{} -> [剩余通道数量] {} -> ", this.getClass().getName(), channelGroup.size());
        } catch (Exception e) {
            log.error("异常关闭时的异常 [{}] ", e.getCause());
        }

    }


    /**
     * 数据下发操作
     *
     * @param frameAndChannel
     */
    @RabbitListener(queues = "${topic.rabbit.gatewayWriteChannel}")
    @RabbitHandler
    public void dataDistribution(Map<String, Object> frameAndChannel) {
        log.info("下发的数据内容:[{}]", frameAndChannel);
        if (frameAndChannel == null) {
            return;
        }
        String channelLongId = (String) frameAndChannel.get("channelLongId");
        if (StringUtils.isEmpty(channelLongId)) {
            return;
        }
        Object gatewayChannelRelation = redisTemplate.opsForValue().get(channelLongId);
        if (gatewayChannelRelation == null) {
            return;
        }
        Map<String, Object> gatewayAndChannelRelation = (Map<String, Object>) gatewayChannelRelation;
        ChannelId channelId = (ChannelId) gatewayAndChannelRelation.get("channelId");
        Channel channel = channelGroup.find(channelId);
        threadPoolService.writeAndFlush(channel, frameAndChannel);
    }
}
